package cn.com.csg.smartenergy.mqtt;

import cn.com.csg.smartenergy.common.Constant;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Data
@Component
public class DeviceOnlineHandler {

//    @Autowired
//    private IotDeviceService deviceService;
//
//    @Autowired
//    private PowerDeviceOnlineService deviceOnlineService;
//
//    @Autowired
//    private RedisService redisService;

    private AbstractMqttClient client;

    /**
     * APP上线通知接口topic：@app/notify/event/gwTerminal/status/@manufacturerId/@manufacturerName/@deviceType/@model/@protocolType/@nodeId
     * app: app名称
     * manufacturerId: 终端厂家ID
     * manufacturerName: 厂家名称
     * deviceType: 终端类型
     * model: 终端型号
     * protocolType: 协议类型
     * nodeId: 终端序列号，必须唯一
     */
    public static final String DEVICE_STATUS = "/notify/event/gwTerminal/status/";

    public static final String PUBLISH_DEVICE_STATUS = "/device/state/";

    /**
     * IOT平台的返回连接状态接口topic：esdk/notify/event/terminal/connStatus/@manufacturerId/@manufacturerName/@deviceType/@model/@nodeId 参数同上
     */
    public static final String IOT_STATUS = "esdk/notify/event/terminal/connStatus/";

    @Value("${real.time.data.prefix}")
    private String prefix;

    @Value("${device.heart.miss}")
    private Integer miss;

    /**
     * 上线消息处理回调异步方法
     *
     * @param topic
     * @param message
     * @param client
     * @throws MqttException
     */
    @Async
    public void handleOnlineMessage(String topic, MqttMessage message, AbstractMqttClient client) throws MqttException {
        System.out.println("上线消息处理回调异步方法handleOnlineMessage");
        this.client = client;
        Long deviceId;
        String msgPayload = new String(message.getPayload(), StandardCharsets.UTF_8);
        log.info(String.format("设备上下线：Topic: %s, Payload: %s", topic, msgPayload));
//        if (topic.contains(DEVICE_STATUS)) {
//            String[] topics = topic.split(Constant.FORWARD_SLASH);
//            String clientId = new StringBuilder()
//                    .append(topics[5]).append(Constant.UNDER_LINE)
//                    .append(topics[7]).append(Constant.UNDER_LINE)
//                    .append(topics[8]).append(Constant.UNDER_LINE)
//                    .append(topics[10]).toString();
//            // 验证设备是否存在
//            if (DeviceMqttOnlineClient.DEVICE_IDS.containsKey(clientId)) {
//                deviceId = DeviceMqttOnlineClient.DEVICE_IDS.get(clientId);
//            } else {
//                IotDeviceEntity params = new IotDeviceEntity();
//                params.setManufacturerId(topics[5]);
//                params.setDeviceType(topics[7]);
//                params.setModel(topics[8]);
//                params.setNodeId(topics[10]);
//                IotDeviceEntity device = this.deviceService.getDeviceInfo(params);
//                if (device == null) {
//                    log.warn(String.format("数据库中找不到设备信息：ManufacturerId: %s, DeviceType: %s, Model: %s, NodeId: %s", topics[5], topics[7], topics[8], topics[10]));
//                    return;
//                }
//                deviceId = device.getId();
//                DeviceMqttOnlineClient.DEVICE_IDS.put(clientId, deviceId);
//            }
//            // 设备上线状态
//            if (DeviceMqttOnlineClient.LAST_BEAT_TIME.containsKey(clientId)) {
//                Long lastTime = DeviceMqttOnlineClient.LAST_BEAT_TIME.get(clientId);
//                // 如果两次报文时间小于30秒，实际上线报文是1秒一次
//                if (System.currentTimeMillis() - lastTime < 30 * 1000) {
//                    message.setQos(1);
//                    this.replayOnlineMessage(message, topics);
//                }
//            } else {
//                message.setQos(1);
//                this.replayOnlineMessage(message, topics);
//            }
//            DeviceMqttOnlineClient.LAST_BEAT_TIME.put(clientId, System.currentTimeMillis());
//            // 更新设备状态逻辑处理
////            String msgPayload = new String(message.getPayload(), StandardCharsets.UTF_8);
//            JSONObject payload = JSON.parseObject(msgPayload);
//            if (payload.get(Constant.MSG_BODY) != null) {
//                JSONObject body = payload.getJSONObject(Constant.MSG_BODY);
//                // 获取设备上传状态
//                String newState = body.getString(Constant.STATE);
//                if (!StringUtils.isEmpty(newState)) {
//                    String upperNewState = newState.toUpperCase();
//                    body.put(Constant.STATE, upperNewState);
//                    String key = prefix + "_state_" + deviceId+"_type_"+topics[7];
//                    // 获取 redis 中的设备状态：online/offline
//                    Object oldState = redisService.hGet(key, Constant.STATE);
//                    // 更新redis中的状态和时间
//                    redisService.hPut(key, Constant.STATE, upperNewState);
//                    redisService.hPut(key, Constant.EVENT_TIME, body.getString(Constant.EVENT_TIME));
//                    // 设置五分钟过期时间，第五次心跳不来时过期，预留5s的缓冲时间，防止设备第五次心跳上线
//                    redisService.expire(key, miss*60 + 5, TimeUnit.SECONDS);
//                    // 状态改变时更新数据库，并推送到页面
//                    if (oldState == null || !upperNewState.equals(oldState.toString())) {
//                        // 存储设备状态
//                        IotDeviceEntity deviceEntity = new IotDeviceEntity();
//                        deviceEntity.setId(deviceId);
//                        // 设备最新状态
//                        deviceEntity.setStatus(upperNewState);
//                        deviceEntity.setDeviceType(topics[7]);
//                        // 保存设备状态到数据库
////                        log.info(String.format("保存设备状态：deviceId: %d, state: %s", deviceId, upperNewState));
//                        String deviceStateJson = JSON.toJSONString(deviceEntity);
//                        redisService.lRightPush(prefix.toUpperCase() + "_" + Constant.QUEUE_DEVICE_STATE, deviceStateJson);
////                        this.deviceService.updateById(deviceEntity);
//                        String stateTopic = prefix + PUBLISH_DEVICE_STATUS + deviceId;
//                        MqttMessage stateMessage = new MqttMessage();
//                        stateMessage.setQos(1);
//                        body.put(Constant.DEVICE_ID, deviceId);
//                        stateMessage.setPayload(body.toJSONString().getBytes());
//                        this.publishOnlineMessage(stateMessage, stateTopic);
//
//                        if ("ONLINE".equals(upperNewState)) {
//                            // 设备上线
//                            PowerDeviceOnlineEntity onlineEntity = new PowerDeviceOnlineEntity();
//                            onlineEntity.setBegtime(new Date());
//                            onlineEntity.setDeviceId(deviceId);
//                            onlineEntity.setDeviceType(topics[7]);
////                            log.info(String.format("保存设备上线记录：deviceId: %d, state: %s", deviceId, upperNewState));
//                            String onlineJson = JSON.toJSONString(onlineEntity);
//                            redisService.lRightPush(prefix.toUpperCase() + "_" + Constant.QUEUE_ONLINE, onlineJson);
//                            // 删除设备异常记录
////                            this.deviceService.deleteErrorOnlineRecord(deviceId);
////                            this.deviceService.saveOrUpdateOnlineRecord(onlineEntity);
//
//                        } else {
//                            // 设备下线
//                            IotDeviceEntity deviceOffline = new IotDeviceEntity();
//                            deviceOffline.setId(deviceId);
//                            deviceOffline.setStatus(Constant.OFFLINE);
//                            deviceOffline.setDeviceType(topics[7]);
//                            // 更新设备下线时间
//                            String offlineJson = JSON.toJSONString(deviceOffline);
//                            redisService.lRightPush(prefix.toUpperCase() + "_" + Constant.QUEUE_OFFLINE, offlineJson);
//                        }
//                    }
//                }
//            }
//        } else {
//            // 设备建立连接
//            JSONObject payload = JSON.parseObject(new String(message.getPayload(), StandardCharsets.UTF_8));
//            if (!topic.contains("disconnected")) {// 连接上来了
//                log.info(String.format("设备建立连接：Topic: %s, Payload: %s", topic, payload));
//            } else {// 连接断开了
//                log.info(String.format("设备断开连接：Topic: %s, Payload: %s", topic, payload));
//            }
//        }

    }

    /**
     * 处理设备上线队列
     */
    @Async
    public void handleOnlineQueue() {
        System.out.println("处理设备上线队列handleOnlineQueue");
//        while(true) {
//            try {
//                Thread.sleep(2000);
//                List<String> deviceList = redisService.lRange(prefix.toUpperCase() + "_" + Constant.QUEUE_DEVICE_STATE, 0, 4999);
//                if (deviceList != null && !deviceList.isEmpty()) {
//                    redisService.lTrim(prefix.toUpperCase() + "_" + Constant.QUEUE_DEVICE_STATE, deviceList.size(), -1);
//                    List<IotDeviceEntity> deviceEntityList = new ArrayList<>();
//                    for (String device : deviceList) {
//                        IotDeviceEntity entity = JSON.parseObject(device, IotDeviceEntity.class);
//                        //新加的，如果是高低开关柜，则开关柜下面的开关都为上线状态
//                        if(StringUtils.equals("LVS",entity.getDeviceType())){
//                            List<IotDeviceEntity> childList = this.deviceService.getBaseMapper().selectList(
//                                    new LambdaQueryWrapper<IotDeviceEntity>().eq(IotDeviceEntity::getParentId, entity.getId()));
//                            if(CollectionUtils.isNotEmpty(childList)){
//                                for(IotDeviceEntity childEntity:childList){
//                                    childEntity.setStatus(entity.getStatus());
//                                    deviceEntityList.add(childEntity);
//                                }
//                            }
//                        }
//                        deviceEntityList.add(entity);
//                    }
//                    this.deviceService.updateBatchById(deviceEntityList);
//                    log.info(String.format("批量更新设备上线状态：个数: %d", deviceList.size()));
//                }
//                List<String> onlineList = redisService.lRange(prefix.toUpperCase() + "_" + Constant.QUEUE_ONLINE, 0, 4999);
//                if (onlineList != null && !onlineList.isEmpty()) {
//                    redisService.lTrim(prefix.toUpperCase() + "_" + Constant.QUEUE_ONLINE, onlineList.size(), -1);
//                    List<PowerDeviceOnlineEntity> onlineEntityList = new ArrayList<>();
//                    for (String online : onlineList) {
//                        onlineEntityList.add(JSON.parseObject(online, PowerDeviceOnlineEntity.class));
//                    }
//                    // 批量删除所有的异常设备数据
//                    this.deviceOnlineService.getBaseMapper().deleteErrorOnlineRecords(onlineEntityList);
//                    // 保存所有设备上线记录
//                    this.deviceOnlineService.saveOrUpdateBatch(onlineEntityList);
//                    log.info(String.format("批量插入设备上线记录：个数: %d", onlineList.size()));
//                }
//            } catch (Exception e) {
//                log.error(String.format("设备上线入库时出现异常"), e);
//            }
//        }
    }


    /**
     * 用于回复设备上线信息
     *
     * @param message 消息
     * @param topics 解析后topic信息
     * @throws MqttException
     */
    public void replayOnlineMessage(MqttMessage message, String[] topics) throws MqttException {
        System.out.println("用于回复设备上线信息");
        String publishTopic = new StringBuilder(IOT_STATUS)
                .append(topics[5]).append(Constant.FORWARD_SLASH)
                .append(topics[6]).append(Constant.FORWARD_SLASH)
                .append(topics[7]).append(Constant.FORWARD_SLASH)
                .append(topics[8]).append(Constant.FORWARD_SLASH)
                .append(topics[10]).toString();
//        JSONObject payload = JSON.parseObject(new String(message.getPayload(), StandardCharsets.UTF_8));
        log.info(String.format("回复设备上下线：Topic: %s", publishTopic));
        this.client.publish(publishTopic, message);
    }

    /**
     * 用于Websocket推送设备上线状态变化
     *
     * @param message 消息
     * @param topic topic信息
     * @throws MqttException
     */
    public void publishOnlineMessage(MqttMessage message, String topic) throws MqttException {
//        JSONObject payload = JSON.parseObject(new String(message.getPayload(), StandardCharsets.UTF_8));
        log.info(String.format("设备状态发生变化：Topic: %s", topic));
        this.client.publish(topic, message);
    }
    public static void main(String[] args) {
        JSONObject payload = JSON.parseObject("{\"body\": {\"state\": \"online\", \"event-time\": \"2021-04-15T06:32:11Z\"}, \"timestamp\": \"2021-04-15T06:32:11Z\"}");
    }
}
